package com.atguigu.flink.datastreamapi.source;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.datagen.DataGenerator;
import org.apache.flink.streaming.api.functions.source.datagen.DataGeneratorSource;
import org.apache.flink.streaming.api.functions.source.datagen.RandomGenerator;

/**
 * Created by Smexy on 2023/4/4
 *
 *      数据生成器是flink提供的一个快速模拟测试数据的数据源。
 *          模拟WaterSensor数据为例。
 */
public class Demo4_DataGen
{
    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 3333);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);

        env.setParallelism(6);

        /*
            数据生成器目前还是使用老的api，并没有新的api

            DataGeneratorSource(DataGenerator<T> generator) {
        this(generator, Long.MAX_VALUE, null)
        }
             DataGeneratorSource(
            DataGenerator<T> generator, //生成数据
            long rowsPerSecond, //每秒生成多少条,总行数。
                                  实际上每个Source算子的实例生成的行数  ceil(rowsPerSecond / Parallelism)
            @Nullable Long numberOfRows //一共要生成多少条，如果是无限，传null。不是null，就是有界流)
         */
        DataGeneratorSource dataGeneratorSource = new DataGeneratorSource(new RandomGenerator<WaterSensor>()
        {
            /*
                构造数据
                    模拟id ： s1---s10
                    vc:     1000-3000
             */
            @Override
            public WaterSensor next() {
                return new WaterSensor(
                    "s" + RandomUtils.nextInt(1, 11),
                    System.currentTimeMillis(),
                    RandomUtils.nextInt(1000, 3001)
                );
            }
        },
            4,
            null);

        /*
            把数据源加入到环境

            是一个可以并行运行的Source
         */
        env.addSource(dataGeneratorSource)
           .returns(WaterSensor.class)
           .print();



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }



    }
}
